package com.taobao.analyse.reduce;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.taobao.analyse.constants.SplitConstants;

public class MergeResultReducer extends MapReduceBase implements
		Reducer<LongWritable, Text, Text, Text> {

	@Override
	public void reduce(LongWritable key, Iterator<Text> values,
			OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
		Map<String, Integer> updatePriceCount = new HashMap<String, Integer>();
		Map<String, Integer> updateCount = new HashMap<String, Integer>();
		String categoryName = "";
		while (values.hasNext()) {
			String info = values.next().toString();
			if(info.startsWith("categoryName")){
				categoryName = info.split(SplitConstants.SPLIT_AT)[1];
			}else{
				updateCountInfo(info, updateCount, updatePriceCount);
			}
		}
		
		Set<String> dates = updateCount.keySet();
		StringBuilder sb_dateList = new StringBuilder();
		StringBuilder sb_changeRate = new StringBuilder();
		DecimalFormat  df =new DecimalFormat("0.00000");  
		double priceChangeCount = 0;
		double totalEditCount = 0;
		for(String date : dates){
			sb_dateList.append(date).append(SplitConstants.SPLIT_OUTPUT);
			priceChangeCount = updatePriceCount.get(date);
			totalEditCount = updateCount.get(date);
			sb_changeRate.append(df.format(priceChangeCount/totalEditCount)).append(SplitConstants.SPLIT_OUTPUT);
		}
		try{
			output.collect(new Text(key+"_"+categoryName), new Text(sb_dateList.toString()));
			output.collect(new Text(key+"_"+categoryName), new Text(sb_changeRate.toString()));
		}catch(Exception e){
			//do nothing
		}
	}
	
	private void updateCountInfo(String data, Map<String, Integer> updateCount,
			Map<String, Integer> updatePriceCount) {
		
		if(StringUtils.isEmpty(data) || data.startsWith("time")){
			return;
		}
		
		//String[] info = data.split(SplitConstants.SPLIT_AT);
		String[] countList = data.split(SplitConstants.SPLIT_POUND);
		// 2012-04-03:0_1#2012-04-04:0_1#
		for (String list : countList) {
			String[] updataInfo = list.split(SplitConstants.SPLIT_COLON);
			// updataInfo[1] -> 0_1
			String[] changed_total = updataInfo[1]
					.split(SplitConstants.SPLIT_UNDER_BAR);
			if (updateCount.get(updataInfo[0]) == null) {
				updateCount.put(updataInfo[0],
						Integer.valueOf(changed_total[1]));
				updatePriceCount.put(updataInfo[0],
						Integer.valueOf(changed_total[0]));
			} else {
				//进行累加操作
				updateCount.put(
						updataInfo[0], 
						updateCount.get(updataInfo[0]) + Integer.valueOf(changed_total[1]));
				updatePriceCount.put(
						updataInfo[0],
						updatePriceCount.get(updataInfo[0]) + Integer.valueOf(changed_total[0]));
			}
		}

	}

}
